package com.gitee.sidihuo.sse;

import com.gitee.sidihuo.sse.base.SseEmitterUtf8;
import com.gitee.sidihuo.sse.base.SseThreadPool;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;

/**
 * 流式响应Http接口SSE所有会话上下文容器
 */
@Slf4j
public class SseContextHolder {

    /**
     * 单例对象
     */
    private static SseContextHolder sseContextHolder = new SseContextHolder();
    /**
     * 容器
     */
    private static Map<String, SseEmitter> sseMap = new ConcurrentHashMap<>();
    /**
     * 异步响应任务线程池
     */
    private ExecutorService executorService = new SseThreadPool(5, 100);

    /**
     * 自动清理map 避免异常情况内存泄漏
     */
    private boolean autoClearSseMap;

    /**
     * 单例私有构造函数
     */
    private SseContextHolder() {
        log.info("SseContextHolder construct ...");
    }

    public static SseContextHolder getInstance() {
        return sseContextHolder;
    }

    /**
     * servlet 收到请求后需要立即注册
     *
     * @param ssePara 参数
     */
    public void subscribe(SsePara ssePara) {
        if (ssePara == null || ssePara.getSseEmitter() == null || StringUtils.isBlank(ssePara.getId())) {
            throw new RuntimeException("subscribe SsePara is illegal");
        }
        SseEmitter sseEmitter = ssePara.getSseEmitter();
        String id = ssePara.getId();
        if (sseMap.containsKey(id)) {
            log.info("sse already subscribed {} need resubscribe", id);
            complete(id);
        }
        sseMap.put(id, sseEmitter);
        sseEmitter.onTimeout(() -> {
            complete(id);
            log.warn("超时回调通知 sse is timeout and removed {}", id);
        });
        sseEmitter.onError((t) -> {
            complete(id);
            if ((t instanceof IOException) && ("Broken pipe".equalsIgnoreCase(t.getMessage()))) {
                log.warn("异常回调通知 sse is onError and removed , IOException maybe client disconnect {}", id);
            } else {
                log.warn("异常回调通知 sse is onError and removed {}", id, t);
            }
        });
        sseEmitter.onCompletion(() -> {
            if (sseMap.containsKey(id)) {
                sseMap.remove(id);
                log.warn("sse is already completed but in context map and removed {}", id);
            }
            log.info("结束回调通知 sse is already completed and confirm removed {}", id);
        });
        int size = sseMap.size();
        log.info("sse subscribe success {} current sse clients num = {}   jarCompileTime {}", id, size, SseJarCompiler.compileTime);

        //清理过多的链接避免内存泄漏
        try {
            clearExpired(ssePara, size);
        } catch (Throwable t) {
            log.warn("clearExpired Throwable {}", id, t);
        }

    }

    /**
     * sse推送消息内容
     *
     * @param id      会话id
     * @param message 参数
     * @throws IOException IO
     */
    public void pushMessage(String id, String message) throws IOException {
        SseEmitter sseEmitter = sseMap.get(id);
        if (sseEmitter == null) {
            log.warn("sse not exist cannot push message {} {}", id, message);
            return;
        }
        log.info("sse push message {} {}", id, message);
        try {
            sseEmitter.send(message);
        } catch (IOException e) {
            String ioExceptionMsg = e.getMessage();
            Throwable cause = e.getCause();
            if ("java.io.IOException: Broken pipe".equalsIgnoreCase(ioExceptionMsg)
                    && cause != null && StringUtils.equalsIgnoreCase(cause.getMessage(), "Broken pipe")) {
                throw new RuntimeException("IOException maybe client disconnect " + id);
            }
            throw e;
        }
    }

    /**
     * 主动关闭sse结束http连接
     *
     * @param id 会话id
     */
    public void complete(String id) {
        SseEmitter sseEmitter = sseMap.get(id);
        if (sseEmitter == null) {
            log.warn("sse not exist no need complete {}", id);
            return;
        }
        log.info("sse complete {}", id);
        sseMap.remove(id);
        sseEmitter.complete();
    }

    public ExecutorService getExecutorService() {
        return executorService;
    }

    public SseEmitter getSseEmitter(String id) {
        SseEmitter sseEmitter = sseMap.get(id);
        return sseEmitter;
    }

    /**
     * 修改默认线程池大小
     *
     * @param executorService 参数
     */
    public void setExecutorService(ExecutorService executorService) {
        this.executorService = executorService;
    }


    public void setAutoClearSseMap(boolean autoClearSseMap) {
        this.autoClearSseMap = autoClearSseMap;
    }

    private void clearExpired(SsePara ssePara, int size) {
        if (!autoClearSseMap) {
            return;
        }
        if (size < 1000) {
            return;
        }
        int mode = ssePara.hashCode() % 1000;
        if (mode > 5) {
            return;
        }
        long currentTimeMillis = System.currentTimeMillis();
        Iterator<Map.Entry<String, SseEmitter>> iterator = sseMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, SseEmitter> next = iterator.next();
            SseEmitter value = next.getValue();
            if (!(value instanceof SseEmitterUtf8)) {
                continue;
            }
            long liveTime = currentTimeMillis - ((SseEmitterUtf8) value).getCreateTime();
            Long timeout = value.getTimeout();
            if (timeout == 0L) {
                if (liveTime > 36000000) {
                    log.warn("SseEmitterUtf8 [forever] already alive {} {}", liveTime, next.getKey());
                }
            } else {
                if (liveTime > (timeout * 2)) {
                    log.warn("SseEmitterUtf8 set live time {} but already alive {} and auto removed {}", liveTime, liveTime, next.getKey());
                    iterator.remove();
                }
            }

        }
    }

}
